package com.hujf.flink.selfFunction;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * @Author hujf
 * @Date 2022/4/18 11:10
 * @Description TODO
 */
public class FuncTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment =StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> integerDataStreamSource = environment.fromElements("hello how are you my name is hujf and how about you");


        //map 的过程
        DataStream<Tuple2<String,Integer>> map = integerDataStreamSource.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (s, out) -> {
            for(String word: s.split(" ")){
                out.collect(new Tuple2<>(word,1));
            }
        }).returns(TupleTypeInfo.getBasicTupleTypeInfo(String.class, Integer.class));
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = map.keyBy(value -> value.f0).sum(1);

        sum.print();

        environment.execute();

    }
}
